package com.atguigu.gmall.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.DruidDSUtil;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;

/**
 * @author Felix
 * @date 2022/7/29
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private DruidDataSource dataSource;

    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidDSUtil.createDataSource();
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        // {"tm_name":"华为","sink_table":"dim_base_trademark","id":3}
        //获取目的地表名
        String sinkTable = jsonObj.getString("sink_table");
        //将sink_table属性从jsonObj中删除掉 {"tm_name":"华为","id":3}
        jsonObj.remove("sink_table");

        //获取操作类型
        String type = jsonObj.getString("type");
        jsonObj.remove("type");


        //拼接upsert语句
        String upsertSql = "upsert into " + GmallConfig.PHOENIX_SCHEMA + "." + sinkTable
            + " (" + StringUtils.join(jsonObj.keySet(), ",") + ") " +
            " values('" + StringUtils.join(jsonObj.values(), "','") + "')";

        System.out.println("向phoenix表中插入数据的SQL:" + upsertSql);


        //建立连接
        Connection conn = dataSource.getConnection();
        PhoenixUtil.executeSql(upsertSql,conn);

        //如果当前执行的是update操作，那么将缓存中的维度数据删除掉
        if("update".equals(type)){
            DimUtil.delCached(sinkTable,jsonObj.getString("id"));
        }
    }
}
